package cas.ihep.hadoop.rand;

import cas.ihep.hadoop.util.Utils;
import cas.ihep.util.RandomGenerator;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.io.PrintWriter;
import java.util.Random;

public class Rand extends Configured implements Tool {

    private static final Log LOG = LogFactory.getLog(Rand.class);

    private static final Path TMP_DIR=new Path("/tmp/input/"+Rand.class.getSimpleName());

    @Override
    public int run(String[] strings) throws Exception {
        Configuration jobconf=getConf();
        FileSystem fs=FileSystem.get(jobconf);
        String jobName = jobconf.get("ybj.jobName", RandomGenerator.instance.next(Rand.class));
        final Path inDir=new Path(TMP_DIR,jobName);
        jobconf.set(Utils.TmpInputDirKey,inDir.toUri().toString());
        if (fs.exists(inDir)) {
            throw new IOException("Tmp directory " + fs.makeQualified(inDir)
                    + " already exists.  Please remove it or change your Job Name!.");
        }
        if (!fs.mkdirs(inDir)) {
            throw new IOException("Cannot create input directory " + inDir);
        }
        int count=Integer.parseInt(strings[0]);
        Random rand=new Random();
        for(int i=0;i<count;i++){
            String s=Integer.toString(i);
            final Path file = new Path(inDir, s+".txt");
            try(PrintWriter out = new PrintWriter(fs.create(file,(short)1))){
                out.print(s);
                out.print(' ');
                out.print(300*1024*1024+rand.nextInt(100*1024*1024));
            }
            LOG.info("Wrote input for I #"+i);
        }
        Job job = Job.getInstance(jobconf);
        //job.setJobName("Ybj");
        job.setJarByClass(Rand.class);
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(NullOutputFormat.class);
        job.setMapperClass(RandMapper.class);
        job.setReducerClass(RandReducer.class);
        job.setNumReduceTasks(1);
        //setup input/output directories
        FileInputFormat.setInputPaths(job, inDir);
        job.submit();
        job.waitForCompletion(true);
        return 0;
    }

    public static void main(String args[]) throws Exception {
        System.exit(ToolRunner.run(new Configuration(),new Rand(),args));
    }
}
